package com.example.demo.messager;

import com.example.demo.competitor.GatewayCompetitor;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PropertiesLoaderUtils;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Properties;


/**
 * 处理Mq消息, 对消息不做任何业务操作, 只负责接收消息
 * Created by guoyangqiao on 10/05/2017.
 */
public class MessageReceiver {
    private static final Logger LOGGER = LoggerFactory.getLogger(MessageReceiver.class);
    public static boolean isAlive = false;

    public static final KafkaConsumer<String, String> KAFKA_CONSUMER;
    private static Properties CONFIG = null;

    static{
        Resource resource = new ClassPathResource("/application.properties");
        try {
            CONFIG = PropertiesLoaderUtils.loadProperties(resource);

            Object parallelism = CONFIG.get("STREAM.DEFAULT.PARALLELISM");
            if (parallelism == null)
                LOGGER.warn("没有设置默认流的并行量,将由系统默认赋予");
            else
                System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", (String) parallelism);

            LOGGER.info("获取到配置信息:\n" + CONFIG.entrySet().stream().map(x -> "\t" + x.getKey() + "=" + x.getValue() + "\n").reduce("", (o, n) -> o + n));
            LOGGER.info("当前系统信息:\n" + System.getProperties().entrySet().stream().map(x -> "\t" + x.getKey() + "=" + x.getValue() + "\n").reduce("", (o, n) -> o + n));

        } catch (IOException e) {
            LOGGER.error("没找到这个文件??- -??", e);
            CONFIG = null;
        }


        if (CONFIG == null) {
            KAFKA_CONSUMER = null;
            LOGGER.info("不能初始化kafka");
        }else {

            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,CONFIG.get("KAFKA.BOOTSTRAP.SERVERS"));
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");
            props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
            props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"30000");

            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

            KAFKA_CONSUMER = new KafkaConsumer<String, String>(props);
            LOGGER.info("kafka消费端初始化结束，配置信息：{}",props);
        }

    }

    public static void handler() {
        try {
            isAlive = true;
            while (true) {
                try {
                    List<Object> obj = Collections.singletonList(CONFIG.get("KAFKA.PRODUCER.TOPIC"));
                    //KAFKA_CONSUMER.subscribe(obj);
                    ConsumerRecords<String, String> records = KAFKA_CONSUMER.poll(1000);
                    //TODO:下面的业务。。。？

                    //compete Gateway Then Notify Subsequent If Success 竞争网关然后通知后续如果成功
                    GatewayCompetitor.competeGatewayThenNotifySubsequentIfSuccess(10);
                } catch (Exception e) {
                    LOGGER.error("MQ消息处理异常", e);
                }
            }
        } catch (Exception e) {
            LOGGER.error("MQ异常", e);
        } finally {
//            releaseResources(channel, connection);
            isAlive = false;
        }
    }
}
